[レポート]AWS Lambda Powertoolsを活用してサーバーレスAPIの開発を加速するワークショップに参加しました #SVS306 #AWSreInvent

AWS re:Invent 2024で行われたワークショップ形式のセッション「SVS306: Accelerate development with AWS Lambda Powertools for serverless APIs」に参加しましたので、そのレポートをお届けします。

AWS re:Invent 2024

#AWS

塚本太朗

2024.12.04

はじめに

こんにちは、リテールアプリ共創部の塚本です。
AWS re:Invent 2024で行われたワークショップ形式のセッション「SVS306: Accelerate development with AWS Lambda Powertools for serverless APIs」に参加しましたので、そのレポートをお届けします。

初めてのWorkshop形式のイベント参加で勝手がわからないところがありましたが、ほとんど手順通りに終えられました。
また、英語で質問できたのは良い経験でした。
ワークショップ自体は全部で3段階で構成されていましたが、時間内に2つ目のステップまでしか終わりませんでした。

セッション情報

セッションID: SVS306
タイトル: Accelerate development with AWS Lambda Powertools for serverless APIs
レベル: 300 - Advanced
所要時間: 120分
対象者:
- DevOps Engineer
- Developer / Engineer
- IT Professional
- Technical Manager

セッション概要

IMG_3929（大）.jpeg

[セッションの公式説明]

In this workshop, start with an existing application built with Python and progressively improve your API event handler using Powertools for AWS Lambda. Learn how to implement request and response validation, dynamic routing, exception handling, middleware, and OpenAPI schema generation. Discover how to improve your API event handler with serverless best practices using Python that you can easily extend to other Powertools runtimes. You must bring your laptop to participate.

[日本語訳]

このワークショップでは、Pythonで構築された既存のアプリケーションから始めて、AWS Lambda Powertoolsを使用してAPIイベントハンドラーを段階的に改善していきます。リクエストとレスポンスの検証、動的ルーティング、例外処理、ミドルウェア、OpenAPIスキーマ生成の実装方法を学びます。Pythonを使用したサーバーレスのベストプラクティスでAPIイベントハンドラーを改善する方法を発見し、これを他のPowertools実行環境にも簡単に拡張できます。参加にはラップトップの持参が必要です。

ワークショップの構成

Create your first API
- What is a Lambda integration
- Understanding the legacy application
- Adding Powertools for AWS Lambda (Python)
- Defining routes with Powertools 4.1. Using the Router Object for better organization
- Using different HTTP methods
- Using the Response object for consistent API responses
- Adding CORS protection
- Handling exceptions and not found routes
- Observing your API
- Securing your API
Middleware, Data Validation & Idempotency
- Using Middleware
- Using Data validation
- Ensuring idempotency
OpenApi

「Create your first API」の概要

AWS Lambda Powertools を利用していない実装から始め、徐々に改善していく内容になっています。
例えば、Lambda Powertoolsの以下の機能を利用します。

Router: APIGatewayからのリクエストをパスやメソッドに基づいてルーティングする。以下のような実装。

@router.delete("/orders/<order_id>")
def delete_order(order_id: str):
    orders_table.delete_item(Key={'orderId': order_id})
    
    return Response(
        status_code=HTTPStatus.NO_CONTENT.value,  # HTTP CODE 204
        content_type=content_types.APPLICATION_JSON,
    )

Response: APIGatewayへ返却する形式のレスポンスを簡単に生成できる

return Response(
    status_code=HTTPStatus.OK.value,  # HTTP CODE 200
    content_type=content_types.APPLICATION_JSON,
    body=response['Item'],
)

CORSConfig: CORS周りの設定ができる

cors_config = CORSConfig(allow_origin="https://www.amazon.com", max_age=300)
app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)

metric: CloudWatchメトリクスで統計を確認できる

@metrics.log_metrics(capture_cold_start=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

tracer: X-Rayのトレースに詳細な情報を追加する

tracer = Tracer()
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

「Middleware, Data Validation & Idempotency」の概要

Middlewareを使用し、API実行ができるユーザーを制限していきます。
Middlewareはデコレータとして利用することで、毎回のAPI実行の前に処理を走らせることができます。
今回のワークセッションでは、Middlewareはデコレータとして利用し、特定のユーザーのみがAPI実行できるケースを作成していきました。

ミドルウェアを定義する箇所の実装

app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)
app.use(middlewares=[restrict_data_access]) # ⭐️ middlewareを設定

ミドルウェア自体の実装

def restrict_data_access(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    if "authorization" in app.current_event.headers:
        # Decode the token
        token = app.current_event.headers["authorization"]
        decoded_token = json.loads(base64.b64decode(token).decode('utf-8'))
        #===省略===#

    response = next_middleware(app)
    return response

初めてワークショップに参加した感想

普段TypeScriptで AWS Lambda Powertools を利用しているので、目新しい機能はなかったですが復習になりました。
また、初めてのワークショップで少し緊張していましたが、思ったよりもワークショップのサイトのみで勧められました。
現地で課金を気にせずにワークショップをできる機会は少ないので、この後もワークショップをどんどん入れてきたいと思います！

最終的な実装（参考）

セッションが終わったところまでですが、以下のような実装になりました。
参考のため載せておきます。

app.py: Lambdaのエントリーポイント

#####
# imports - app.py
#####
from http import HTTPStatus
import json
import base64
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, CORSConfig
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.event_handler.exceptions import NotFoundError
from aws_lambda_powertools import Logger, Metrics, Tracer
from all_routes import router
from middleware import restrict_data_access

#####
# Classes, functions and instances - app.py
#####
logger = Logger()
metrics = Metrics()
tracer = Tracer()

cors_config = CORSConfig(allow_origin="https://www.amazon.com", max_age=300)
app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)
app.use(middlewares=[restrict_data_access])


@app.exception_handler([ValueError, AttributeError])
def handle_invalid_payload(ex: ValueError | AttributeError):
    metadata = {"path": app.current_event.path, "http_method": app.current_event.http_method}
    logger.exception(f"Malformed request: {ex}", metadata=metadata)

    return Response(
        status_code=HTTPStatus.BAD_REQUEST.value,
        content_type=content_types.APPLICATION_JSON,
        body={"message": "Invalid request parameters. Please verify your parameters or payload according to our documentation."}
    )

@app.not_found
def handle_not_found_errors(exc: NotFoundError) -> Response:
    logger.info(f"Route not found: {app.current_event.path}")
    return Response(status_code=HTTPStatus.NOT_FOUND.value, content_type=content_types.TEXT_PLAIN, body="Sorry, I don't exist!")

#####
# Lambda handler - app.py
#####
@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

all_routes.py

#####
# imports - all_routes.py
#####
from http import HTTPStatus
from uuid import uuid4
import boto3
import json
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.event_handler.api_gateway import Router
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)

#####
# Classes, functions and instances - all_routes.py
#####
logger = Logger()
metrics = Metrics()
tracer = Tracer()

dynamodb = boto3.resource('dynamodb')
orders_table = dynamodb.Table('OrdersWorkshop')
router = Router()

#####
# Get all orders method - all_routes.py
#####
@router.get("/orders")
def get_all_orders():
    response = orders_table.scan()
    
    if len(response['Items']) > 0:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Items'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "No orders found"}
        )

#####
# Get order method - all_routes.py
#####
@router.get("/orders/<order_id>")
@tracer.capture_method
def get_order(order_id: str):
    response = orders_table.get_item(Key={'orderId': order_id})
    
    # Logging
    logger.info("Searching an order", order_id=order_id)

    # Adding metric
    metrics.add_dimension(name="order_id", value=order_id)
    metrics.add_metric("OrderSearch", unit=MetricUnit.Count, value=1)

    if 'Item' in response:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Item'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "Order not found"},
        )

#####
# Create order method - all_routes.py
#####
@router.post("/orders")
def create_order():
    body = router.current_event.json_body
    order_id = str(uuid4())
    
    item = {
        'orderId': order_id,
        'customerName': body.get('customerName'),
        "restaurantName": body.get('restaurantName'),
        'orderItems': body.get('orderItems'),
        'orderDate': body.get('orderDate'),
        'orderStatus': 'Pending',
        'restaurantId': body.get('restaurantId'),
    }
    
    orders_table.put_item(Item=item)

    return Response(
        status_code=HTTPStatus.CREATED.value,  # HTTP CODE 201
        content_type=content_types.APPLICATION_JSON,
        headers={"Location": f"/orders/{order_id}"},
        body=item,
    )

#####
# Update order method - all_routes.py
#####
@router.put("/orders/<order_id>")
def update_order(order_id: str):
    body = router.current_event.json_body
    
    response = orders_table.update_item(
        Key={'orderId': order_id},
        UpdateExpression='SET customerName = :name, orderItems = :items, orderDate = :date, orderStatus = :status',
        ExpressionAttributeValues={
            ':name': body.get('customerName'),
            ':items': body.get('orderItems'),
            ':date': body.get('orderDate'),
            ':status': body.get('orderStatus'),
        },
        ReturnValues='ALL_NEW'
    )

    return Response(
        status_code=HTTPStatus.OK.value,  # HTTP CODE 200
        content_type=content_types.APPLICATION_JSON,
        body=response['Attributes'],
    )

#####
# Delete order method - all_routes.py
#####
@router.delete("/orders/<order_id>")
def delete_order(order_id: str):
    orders_table.delete_item(Key={'orderId': order_id})
    
    return Response(
        status_code=HTTPStatus.NO_CONTENT.value,  # HTTP CODE 204
        content_type=content_types.APPLICATION_JSON,
    )

#####
# Get all orders per restaurant method - all_routes.py
#####
@router.get("/orders_per_restaurant/<restaurant_id>")
def get_all_orders_per_restaurant(restaurant_id: str):
  
    response = orders_table.scan(
        FilterExpression='restaurantId = :rid',
        ExpressionAttributeValues={
            ':rid': restaurant_id
        }
    )
    
    if len(response['Items']) > 0:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Items'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "No orders found"},
        )

middleware.py

#####
# imports - middleware.py
#####
from http import HTTPStatus
import json
import base64
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)
from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
from aws_lambda_powertools.utilities.typing import LambdaContext

#####
# Middleware function - middleware.py
#####
def restrict_data_access(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    if "authorization" in app.current_event.headers:
        # Decode the token
        token = app.current_event.headers["authorization"]
        decoded_token = json.loads(base64.b64decode(token).decode('utf-8'))

        # Admin should return early, full privileges
        if decoded_token.get("level") == "admin":
            return next_middleware(app)

        restaurant_id = str(decoded_token.get("restaurant_id"))
        path_params = app.context.get("_route_args", {})

        if "restaurant_id" in path_params:
            if str(path_params["restaurant_id"]) != restaurant_id:
                return Response(
                    status_code=HTTPStatus.FORBIDDEN.value,
                    content_type=content_types.APPLICATION_JSON,
                    body={"message": f"Access denied: You don't have permission for restaurant {path_params['restaurant_id']}"}
                )
            return next_middleware(app)

        response = next_middleware(app)
        
        if str(response.body.get("restaurantId")) != restaurant_id:
            return Response(
                status_code=HTTPStatus.FORBIDDEN.value,
                content_type=content_types.APPLICATION_JSON,
                body={"message": "Access denied: You don't have permission to access this order"}
            )
    
    response = next_middleware(app)
    return response